package com.atguigu.gmall.realtime.utils;

import com.atguigu.gmall.realtime.beans.TransientSink;
import com.atguigu.gmall.realtime.common.GmallConfig;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * Author: Felix
 * Date: 2021/8/11
 * Desc: 操作ClickHouse的工具类
 */
public class ClickHouseUtil {
    public static <T>SinkFunction<T> getJdbcSink(String sql){
        //insert into visitor_stats_0224 values(?,?,?,?,?,?,?,?,?,?,?,?)
        SinkFunction<T> sinkFunction = JdbcSink.<T>sink(
                sql,
                new JdbcStatementBuilder<T>() {
                    //获取流中对象obj的属性值，赋值给问号占位符?    参数 T obj :就是流中一条数据
                    @Override
                    public void accept(PreparedStatement ps, T obj) throws SQLException {
                        //获取流中对象所属类的属性
                        /**
                         * getClass() 获取类
                         * getDeclaredFields() 获取当前对象的所有属性包括私有
                         * getFields() 获取所有属性，包括父类属性
                         */
                        Field[] fields = obj.getClass().getDeclaredFields();
                        //对属性数组 进行遍历
                        int skipNum = 0;
                        for (int i = 0; i < fields.length; i++) {
                            //获取每一个属性对象
                            Field field = fields[i];

                            //判断该属性是否有@TransientSink注解修饰
                            TransientSink transientSink = field.getAnnotation(TransientSink.class);
                            if (transientSink != null) {
                                skipNum++;
                                //break;结束循环
                                continue;
                            }

                            //设置私有属性的访问权限
                            field.setAccessible(true);

                            try {
                                //获取对象的属性值
                                Object fieldValue = field.get(obj);
                                //将属性的值  赋值给问号占位符
                                //jdbc的相关操作,查询结果集的列都是从1开始的
                                //注解的原因要跳过索引所以-skipNum
                                ps.setObject(i + 1 - skipNum, fieldValue);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }
                    }
                },
                new JdbcExecutionOptions.Builder()
                        //每个并行度，批量大小（条）
                        .withBatchSize(5)
                        //设置时间，2秒
                        //.withBatchIntervalMs(2000)
                        //最大重新尝试次数
                        //.withMaxRetries()
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        //驱动
                        .withDriverName("ru.yandex.clickhouse.ClickHouseDriver")
                        //url
                        .withUrl(GmallConfig.CLICKHOUSE_URL)
                        .build()
        );
        return sinkFunction;
    }

}
